package com.gugee.ins.data.collector.mblogger.service.impl;

import com.gugee.common.exception.BizException;
import com.gugee.common.exception.SpiderRetryableException;
import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.common.dal.service.dic.InsDicCategoryDao;
import com.gugee.ins.common.dal.service.dic.InsDicCategoryEnumDao;
import com.gugee.ins.common.dal.service.spider.SpiderSvc;
import com.gugee.ins.common.dal.spider.InsBloggerSpiderClient;
import com.gugee.ins.common.model.mysql.blogger.InsBloggerMain;
import com.gugee.ins.common.model.spider.blogger.InsBloggerByIdResponse;
import com.gugee.ins.common.model.spider.blogger.InsBloggerByUsernameResponse;
import com.gugee.ins.common.model.spider.blogger.InsBloggerUserFromId;
import com.gugee.ins.data.collector.mblogger.config.InsMBloggerConfig;
import com.gugee.ins.data.collector.mblogger.service.MBloggerSvc;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.web.client.ResourceAccessException;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: luolifeng
 * Date: 2020-12-24 16:20
 */
@Slf4j
@Service
public class MBloggerSvcImpl implements MBloggerSvc {

    @Autowired
    InsBloggerSpiderClient insBloggerSpiderClient;
    /**
     * 爬虫接口
     */
    @Autowired
    SpiderSvc spiderSvc;

    @Autowired
    InsBloggerMainDao insBloggerMainDao;
    @Autowired
    InsBloggerDao insBloggerDao;
    @Autowired
    ScheduleTemplate scheduleTemplate;

    @Autowired
    InsMBloggerConfig insMBloggerConfig;

    /**
     *
     */
    @Autowired
    InsDicCategoryEnumDao insDicCategoryEnumDao;

    /**
     *
     */
    @Autowired
    InsDicCategoryDao insDicCategoryDao;

    private ThreadPoolExecutor customPoolMain;
    @Autowired
    @Qualifier("insBloggerMsgBaseInfoProducer")
    private KafkaTemplate<Long, InsBloggerMessage> insBloggerMsgBaseInfoProducer;
    @PostConstruct
    public void init() {
        log.info("MBloggerSvcImpl  init");
        customPoolMain = scheduleTemplate.getCustomThreadPool("insMBlg-upd", insMBloggerConfig.getCorePollSize(), insMBloggerConfig.getCorePollSize(), 0, insMBloggerConfig.getQueueSize(), TimeUnit.MINUTES);
    }


    private void run(InsBloggerMessage insBloggerMessage){
        InsBloggerUserFromId insBlogger =spideInsBloggerMain(insBloggerMessage);
        if(insBlogger==null){
            return;
        }
        //根据username爬取信息 需求：只需要商业账号，且分类为空时，采集username接口
        if( Boolean.TRUE.equals(insBlogger.isBusiness()) && !StringUtils.isEmpty(insBlogger.getUsername()) && StringUtils.isEmpty(insBlogger.getCategory())){
            try {
                InsBloggerByUsernameResponse insBloggerByUsernameResponse = insBloggerSpiderClient.getInsBloggerByUserName(insBlogger.getUsername());
                if(insBloggerByUsernameResponse != null
                        && insBloggerByUsernameResponse.getGraphql() != null
                        && insBloggerByUsernameResponse.getGraphql().getUser() != null){
                    insBlogger.setCategory(insBloggerByUsernameResponse.getGraphql().getUser().getCategoryName());
                    // 如果CategoryName没有就用CategoryEnum
                    if(StringUtils.isEmpty(insBlogger.getCategory())
                            && !StringUtils.isEmpty(insBloggerByUsernameResponse.getGraphql().getUser().getCategoryEnum())){
                        String category = insDicCategoryEnumDao.findCategoryByEnum(insBloggerByUsernameResponse.getGraphql().getUser().getCategoryEnum());
                        if(StringUtils.isEmpty(category)){
                            insBlogger.setCategory(category);
                        }
                    }
                    insBlogger.setBusinessCategoryName(insBloggerByUsernameResponse.getGraphql().getUser().getBusinessCategoryName());
                    insBlogger.setFbid(Long.parseLong(insBloggerByUsernameResponse.getGraphql().getUser().getFbid()));
                }
            }catch (Exception e){
                log.error("spider ins blogger_main by username error, msg :{}",e.getMessage());
            }
        }
        if(!StringUtils.isEmpty(insBlogger.getCategory())){
            String overallCategory = insDicCategoryDao.findOverallCategoryByCategory(insBlogger.getCategory());
            insBlogger.setOverallCategoryName(overallCategory);
        }

        insBloggerMainDao.updateMBlogger(insBlogger);
        insBloggerDao.updateBlogger(insBlogger);
    }

    /**
     * 爬取大播主基本信息
     * @param insBloggerMessage   需更新的播主uid
     */
    private InsBloggerUserFromId spideInsBloggerMain(InsBloggerMessage insBloggerMessage) {

        InsBloggerByIdResponse response = spiderSvc.getInsBloggerByUid(insBloggerMessage.getUid());
        if(!"ok".equals(response.getStatus())){
            log.error("uid接口未能正常返回数据,status={}",response.getStatus());
            return null;
        }
        return response.getUser();

    }

    /**
     * 收录大播主
     * @param insBloggerMessages 消息列表
     */
    @Override
    public void handleCollect(List<InsBloggerMessage> insBloggerMessages){
        List<Long> uidS = new ArrayList<>(insBloggerMessages.size());
        insBloggerMessages.forEach(msg-> uidS.add(msg.getUid()));
        List<InsBloggerMain>  insBloggerMains= insBloggerMainDao.produceInsBloggerMainsFromBlogger(uidS,insMBloggerConfig.getClFollowerCounts());
        if(insBloggerMains==null || insBloggerMains.isEmpty()){
            return;
        }
        insBloggerMainDao.batchInsertBloggerMains(insBloggerMains);
        // 发送优先采集消息
        insBloggerMessages.forEach(this::sendMsg);
    }

    /**
     * 更新大播主基本信息,通过线程池启动并发多处理，在重试次数范围内，对部分原因的失败做发送回队列待下次消费
     * @param insBloggerMessages 消息列表
     */
    @Override
    public void handleUpdate(List<InsBloggerMessage> insBloggerMessages) {

        Collection<Future<?>> futures = new LinkedList<>();
        insBloggerMessages.forEach(
                bloggerMessage -> futures.add(customPoolMain.submit(() -> {
                    try {
                        run(bloggerMessage);

                        // 发送优先采集消息
                        sendMsg(bloggerMessage);

                    } catch(ResourceAccessException | BizException |SpiderRetryableException e){
                        log.error(" 超时-----需重试{}次：{}" ,bloggerMessage.getRetryTimes()-1,e.getMessage());
                        if(bloggerMessage.getRetryTimes()>0){
                            bloggerMessage.setRetryTimes(bloggerMessage.getRetryTimes()-1);
                            insBloggerMsgBaseInfoProducer.send(insMBloggerConfig.getInsMBloggerClOnlineTopic(),bloggerMessage);
                        }
                        if(bloggerMessage.getRetryTimes()==0){
                            log.error(" 重试{}次后依旧失败:uid={}:{}" ,bloggerMessage.getRetryLimitTimes(),bloggerMessage.getUid(),e.getMessage());
                        }

                    }
                    log.debug(" 重试{}次后依旧成功:uid={}" ,bloggerMessage.getRetryLimitTimes(),bloggerMessage.getUid());
                }))
        );
        for (Future<?> future : futures) {
            try {
                future.get();
            }  catch(Exception e) {
                e.printStackTrace();
                log.error(" future .{}" ,e.getMessage());
            }
        }
    }

    /**
     * 发送优先采集消息
     * @param insBloggerMessage insBloggerMessage
     */
    public void sendMsg(InsBloggerMessage insBloggerMessage){
        if(insBloggerMessage.getTypeCode() != InsBloggerMsgTypeCodeEnum.UID_UPDATE_F_BLOGGER_TIMING_MIDDLE.getCode()){
            sendCollectMediaMeg(insBloggerMessage);
//        sendCollectFollowerMeg(insBloggerMessage);
        }
    }

    /**
     * 发送消息
     * @param insBloggerMessage insBloggerMessage
     */
    public void sendCollectMediaMeg(InsBloggerMessage insBloggerMessage){
        // 当前时间
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        insBloggerMessage.setRetryTimes(insMBloggerConfig.getInsMediaPriorityRetryTimes());
        insBloggerMessage.setRetryLimitTimes(insMBloggerConfig.getInsMediaPriorityRetryTimes());
        insBloggerMsgBaseInfoProducer.send(insMBloggerConfig.getInsMediaPriorityTopic(),insBloggerMessage);
    }

    /**
     * 发送消息
     * @param insBloggerMessage insBloggerMessage
     */
    public void sendCollectFollowerMeg(InsBloggerMessage insBloggerMessage){
        // 当前时间
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        insBloggerMessage.setRetryTimes(insMBloggerConfig.getInsBloggerFollowerPriorityRetryTimes());
        insBloggerMessage.setRetryLimitTimes(insMBloggerConfig.getInsBloggerFollowerPriorityRetryTimes());
        insBloggerMsgBaseInfoProducer.send(insMBloggerConfig.getInsBloggerFollowerPriorityTopic(),insBloggerMessage);
    }

//    /**
//     * 发送验证中播主消息给kafka
//     * @param uid
//     */
//    public void sendCheckMidBloggerMsg(long uid){
//        InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
//        insBloggerMessage.setUid(uid);
//        insBloggerMsgBaseInfoProducer.send(insMBloggerConfig.getInsMidBloggerCheckOnlineTopic(),insBloggerMessage);
//    }
}
